package com.gitee.xhs.windowState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * @author BigHead
 * @date 2022/7/8 14:02
 * @description TransactionOneMinuteCounter
 */
public class TransactionOneMinuteCounter implements AggregateFunction<Transaction, TransactionCounterDao, TransactionCounterDao> {
    @Override
    public TransactionCounterDao createAccumulator() {
        return new TransactionCounterDao();
    }

    @Override
    public TransactionCounterDao add(Transaction value, TransactionCounterDao accumulator) {

        if (accumulator.isInit()) {
            accumulator.setInit(false);
            accumulator.setAccountId(value.getAccountId());
            accumulator.setStartTime(value.getTimestamp());
            accumulator.setEndTime(value.getTimestamp());
            accumulator.setSumAmount(value.getAmount());
        } else {
            if (value.getTimestamp() < accumulator.getStartTime()) {
                accumulator.setStartTime(value.getTimestamp());
            }
            if (value.getTimestamp() > accumulator.getEndTime()) {
                accumulator.setEndTime(value.getTimestamp());
            }
            accumulator.setSumAmount(
                    accumulator.getSumAmount() + value.getAmount()
            );
        }

        return accumulator;
    }

    @Override
    public TransactionCounterDao getResult(TransactionCounterDao accumulator) {
        return accumulator;
    }

    @Override
    public TransactionCounterDao merge(TransactionCounterDao a, TransactionCounterDao b) {
        if (a.getAccountId() == b.getAccountId()) {
            if (a.getStartTime() > b.getStartTime()) {
                a.setStartTime(b.getStartTime());
            }
            if (a.getEndTime() < b.getEndTime()) {
                a.setEndTime(b.getEndTime());
            }
            a.setSumAmount(
                    a.getSumAmount() + b.getSumAmount()
            );
        }
        return a;
    }
}
